/**
 * 
 */
package org.perfect.batch.partition;

import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

import org.apache.dubbo.config.annotation.DubboReference;
import org.perfect.batch.BatchConstants;
import org.perfect.batch.api.IBatchService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.partition.support.DefaultStepExecutionAggregator;
import org.springframework.batch.core.partition.support.StepExecutionAggregator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.task.TaskExecutor;

/**
 * @author billmse
 *
 */
public class RemoteStepExecutionAggregator implements StepExecutionAggregator {
	
	@DubboReference(injvm = false, check = false)
	private IBatchService service;

	@Autowired
	private JobExplorer jobExplorer;

	@Autowired
	@Qualifier(BatchConstants.BATCH_TASK_EXECUTOR_NAME)
	private TaskExecutor taskExecutor;
	
	@Autowired
	private DefaultStepExecutionAggregator aggregator;

	private Logger logger = LoggerFactory.getLogger(getClass());

	@Override
	public void aggregate(StepExecution result, Collection<StepExecution> executions) {
		try {
			FutureTask<Collection<StepExecution>> task = createTask(executions);
			taskExecutor.execute(task);
			aggregator.aggregate(result, task.get());
		} catch (InterruptedException | ExecutionException e) {
			if (logger.isErrorEnabled()) {
				logger.error("polling step executions error", e);
			}
			aggregator.aggregate(result, executions);
		}
	}

	protected FutureTask<Collection<StepExecution>> createTask(Collection<StepExecution> executions) {
		return new FutureTask<>(new Callable<Collection<StepExecution>>() {
			
			@Override
			public Collection<StepExecution> call() throws Exception {
				Set<StepExecution> result = new HashSet<>(executions.size());
				for (StepExecution stepExecution : executions) {
					StepExecution update = jobExplorer.getStepExecution(stepExecution.getJobExecutionId(), stepExecution.getId());			
					// You can just stop step execution 
					while (update.getStatus().isRunning()) {
						Thread.sleep(update.getJobParameters().getLong(BatchConstants.BATCH_SLEEP_INTERVAL, BatchConstants.BATCH_SLEEP_INTERVAL_DEFAULT));
						update = jobExplorer.getStepExecution(update.getJobExecutionId(), update.getId());
						if (update.getStatus() == BatchStatus.STARTING) {
							service.executeStep(update.getJobExecutionId(), update.getId());
						}
					}
					if (logger.isInfoEnabled()) {
						logger.info("polling {} ", update);
					}
					result.add(update);
				}
				return result;
			}
			
		});
	}

}
